%% Author: vivimice
%% Created: Jan 13, 2012
%% Description: buffered input module
-module(tio_receiver).

%%
%% Include files
%%
-include("telnet.hrl").

%%
%% Records
%%
-record(receiver, {pid, r_states}).
-record(receiver_state, {socket, raw_buffer, data_queue, option_observer}).

%%
%% Exported Functions
%%
-export([start/3, stop/1]).
-export([init_receiver/1, loop_receive/1]).
-export([init_rs/2, loop_rs/2]).
-export([init_observer/2, loop_observer/2]).
-export([read_byte/1, peek_byte/1]).

%%
%% API Functions
%%

%% read single byte
%% current process will be blocked until data is ready 
read_byte(Receiver) when is_record(Receiver, receiver) ->
	RStates = Receiver#receiver.r_states,
	DataQueue = get_data_queue(RStates),
	case blocking_binary_queue:take(DataQueue, 1) of
		<<Byte>> ->	Byte;
		{error, Reason} -> {error, Reason}
	end.

%% peek single byte
%% current process will be blocked until data is ready 
peek_byte(Receiver) when is_record(Receiver, receiver) ->
	RStates = Receiver#receiver.r_states,
	DataQueue = get_data_queue(RStates),
	case blocking_binary_queue:peek(DataQueue, 1) of
		<<Byte>> ->	Byte;
		{error, Reason} -> {error, Reason}
	end.

%% start receiver
start(Socket, OptionObserveFunc, OptionObserveArgs) when is_function(OptionObserveFunc) ->
	OptionObserver = spawn_link(?MODULE, init_observer, [OptionObserveFunc, OptionObserveArgs]),
	RStates = spawn_link(?MODULE, init_rs, [Socket, OptionObserver]),
	
	ReceiverPid = spawn_link(?MODULE, init_receiver, [RStates]),
	gen_tcp:controlling_process(Socket, ReceiverPid),
	inet:setopts(Socket, [{active, once}]),
	
	Receiver = #receiver{pid=ReceiverPid, r_states=RStates},
	Receiver;

start(_Socket, _OptionObserver, _OptionObserveArgs) ->
	erlang:error(badarg).

%% stop receiver
stop(Receiver) when is_record(Receiver, receiver) ->
	Receiver#receiver.pid ! stop;
stop(_Receiver) ->
	erlang:error(badarg).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% telnet option observer
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
init_observer(OptionObserveFunc, OptionObserveArgs) ->
	loop_observer(OptionObserveFunc, OptionObserveArgs).	

loop_observer(OptionObserveFunc, OptionObserveArgs) ->
	receive
		{option, Command, Option} ->
			apply(OptionObserveFunc, [{option, Command, Option}] ++ OptionObserveArgs),
			loop_observer(OptionObserveFunc, OptionObserveArgs);
		
		{suboption, SubOptionBin} ->
			apply(OptionObserveFunc, [{suboption, SubOptionBin}] ++ OptionObserveArgs),
			loop_observer(OptionObserveFunc, OptionObserveArgs);
		
		stop ->
			erlang:exit(normal)
	end.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% receiver state fsm
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
init_rs(Socket, OptionObserver) ->
	RawBuffer = <<>>,
	RStateRec = #receiver_state{socket=Socket,
								option_observer=OptionObserver,
								raw_buffer = RawBuffer, 
								data_queue = blocking_binary_queue:new()},
	?MODULE:loop_rs(RStateRec, []).

loop_rs(RStateRec, DataObservers) ->
	receive
		{From, get, socket} ->
			From ! {socket, get, RStateRec#receiver_state.socket},
			?MODULE:loop_rs(RStateRec, DataObservers);
		
		{From, get, option_observer} ->
			From ! {option_observer, get, RStateRec#receiver_state.option_observer},
			?MODULE:loop_rs(RStateRec, DataObservers);
		
		{From, get, data_queue} ->
			From ! {data_queue, get, RStateRec#receiver_state.data_queue},
			?MODULE:loop_rs(RStateRec, DataObservers);
		
		{From, append, raw, RawData} ->
			PrevRawBuffer = RStateRec#receiver_state.raw_buffer,
			NewRawBuffer = <<PrevRawBuffer/binary, RawData/binary>>,
			NewRStateRec = RStateRec#receiver_state{raw_buffer=NewRawBuffer},
			From ! {raw, append},
			?MODULE:loop_rs(NewRStateRec, DataObservers);
		
		{From, get, raw, Length} ->
			<<RawData:Length/binary, NewRawBuffer/binary>> = RStateRec#receiver_state.raw_buffer,
			NewRStateRec = RStateRec#receiver_state{raw_buffer=NewRawBuffer},
			From ! {raw, get, RawData},
			?MODULE:loop_rs(NewRStateRec, DataObservers);
		
		{From, get, rawbuf_size} ->
			From ! {rawbuf_size, get, byte_size(RStateRec#receiver_state.raw_buffer)},
			?MODULE:loop_rs(RStateRec, DataObservers);
		
		close ->
			blocking_binary_queue:close(RStateRec#receiver_state.data_queue),
			erlang:exit(normal)
	end.

get_socket(RStates) ->
	RStates ! {self(), get, socket},
	receive
		{socket, get, Socket} ->
			Socket
	end.

get_option_observer(RStates) ->
	RStates ! {self(), get, option_observer},
	receive
		{option_observer, get, OptionObserver} ->
			OptionObserver
	end.

get_data_queue(RStates) ->
	RStates ! {self(), get, data_queue},
	receive
		{data_queue, get, DataQueue} ->
			DataQueue
	end.

append_raw(RStates, RawData) ->
	RStates ! {self(), append, raw, RawData},
	receive
		{raw, append} ->
			ok
	end.

get_raw(RStates, Length) ->
	RStates ! {self(), get, raw, Length},
	receive
		{raw, get, RawData} ->
			RawData
	end.

get_rawbuf_size(RStates) ->
	RStates ! {self(), get, rawbuf_size},
	receive
		{rawbuf_size, get, Size} ->
			Size
	end.

append_data(RStates, Data) ->
	DataQueue = get_data_queue(RStates),
	blocking_binary_queue:put(DataQueue, Data).

close_rstates(RStates) ->
	RStates ! close.
											 
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% receivers fsm
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
init_receiver(RStates) ->
	?MODULE:loop_receive(RStates).

% RState is receiver_state
loop_receive(RStates) ->
	case recv(RStates, 1) of
		{ok, <<?IAC>>} ->
			receive_command(RStates);
		
		{ok, Data} ->
			append_data(RStates, Data),
			?MODULE:loop_receive(RStates)
	end.

receive_command(RStates) ->
	case recv(RStates, 1) of
		
		% read option
		{ok, <<CommandByte>>} when
		  		CommandByte == ?DO; CommandByte == ?DONT;
				CommandByte == ?WILL; CommandByte == ?WONT ->
			case recv(RStates, 1) of
				{ok, <<OptionByte>>} ->
					option_received(get_option_observer(RStates), CommandByte, OptionByte),
					?MODULE:loop_receive(RStates)
			end;
		
		% read escaped IAC
		{ok, <<?IAC>>} ->
			append_data(RStates, <<?IAC>>),
			?MODULE:loop_receive(RStates);
		
		% read suboption
		{ok, <<?SB>>} ->
			case receive_suboption(RStates) of
				{ok, SubOptionBin} ->
					suboption_received(get_option_observer(RStates), SubOptionBin),
					?MODULE:loop_receive(RStates)
			end;
		
		{ok, <<CommandByte>>} ->
			logger:warn("Discarded unrecognized telnet command: ~w", [CommandByte]),
			?MODULE:loop_receive(RStates)
		
	end.

receive_suboption(RStates) ->
	loop_receive_suboption(RStates, <<>>).

loop_receive_suboption(RStates, SubOptionBin) ->
	case recv(RStates, 1) of
		{ok, <<?IAC>>} ->
			loop_receive_suboption_before_end(RStates, SubOptionBin);
		
		{ok, DataBin} ->
			loop_receive_suboption(RStates, <<SubOptionBin/binary, DataBin/binary>>)
	end.

loop_receive_suboption_before_end(RStates, SubOptionBin) ->
	case recv(RStates, 1) of
		{ok, <<?SE>>} ->
			{ok, SubOptionBin};
		
		{ok, DataBin} ->
			loop_receive_suboption(RStates, <<SubOptionBin/binary, ?IAC, DataBin/binary>>)
	end.

option_received(OptionObserver, Command, Option) ->
	OptionObserver ! {option, Command, Option}.

suboption_received(OptionObserver, SubOptionBin) ->
	OptionObserver ! {suboption, SubOptionBin}.

%% receive data from socket, and also process reading request
recv(RStates, Length) ->
	RawBufSize = get_rawbuf_size(RStates),
	if 
		% buf is insufficient
		RawBufSize < Length ->
			Socket = get_socket(RStates),
			receive
				{tcp, Socket, RawData} ->
					append_raw(RStates, RawData),
					inet:setopts(get_socket(RStates), [{active, once}]),
					recv(RStates, Length);
				
				{tcp_closed, Socket} ->
					logger:error("socket closed. receiver exit."),
					receiver_exit(RStates);
				
				{tcp_error, Reason} ->
					logger:error("Error receiving data, Reason: ~p.", [Reason]),
					receiver_exit(RStates);
				
				stop ->
					receiver_exit(RStates)
			end;
		
		% buf contains sufficient data
		true ->
			{ok, get_raw(RStates, Length)}
	end.

receiver_exit(RStates) ->
	get_option_observer(RStates) ! stop,
	close_rstates(RStates),
	erlang:exit(normal).
